
package com.innovamed.websocket.handler;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

import javax.websocket.OnClose;
import javax.websocket.OnError;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import com.innovamed.websocket.constants.RedisConstants;
import com.innovamed.websocket.model.QueueMsg;
import com.innovamed.websocket.model.SessionInfo;
import com.innovamed.websocket.model.WsMsg;
import com.innovamed.websocket.service.RabbitMqService;
import com.innovamed.websocket.utils.redis.RedisUtils;
import com.innovamed.websocket.utils.websocket.WebSocketUtils;

import cn.hutool.json.JSONUtil;


/**
 * Copyright: Copyright (c) 2019 Jun_Zhou
 *
 * @ClassName: SocketServer.java
 * @Description: sockeet服务端逻辑;
 * Attention:前端调用必须指定服务端的具体Ip,eg:ws://localhost:8080/socketServer/20142213;
 * @version: v1.0.0
 * @author: JunZhou
 * @Email: 1769676159@qq.com
 * @Site: CERNO
 * @date: 2019年1月16日 下午2:20:14
 */
@ServerEndpoint(value = "/ws/socketServer/{userId}")
@Component
public class SocketServer
{
	public static RedisUtils redisUtils;
	
	@Autowired
	public void setRedisUtils(RedisUtils redisUtils)
	{
		SocketServer.redisUtils = redisUtils;
	}
	
	public static RabbitMqService rabbitMqService;
	
	@Autowired
	public void setRabbitMqService(RabbitMqService rabbitMqService)
	{
		SocketServer.rabbitMqService = rabbitMqService;
	}
	
	public static RedisTemplate<String, Object> redisTemplate;
	
	@Autowired
	public void setRedisTemplate(RedisTemplate<String, Object> redisTemplate)
	{
		SocketServer.redisTemplate = redisTemplate;
	}
	
	public static WebSocketUtils webSocketUtils;
	
	@Autowired
	public void setWebSocketUtils(WebSocketUtils webSocketUtils)
	{
		SocketServer.webSocketUtils = webSocketUtils;
	}
	
	//全局会话;
	private Session						session;
	//会话池;
	private static Map<String, Session>	sessionPool	= new HashMap<String, Session>();
	//会话Id;
	private static Map<String, String>	sessionIds	= new HashMap<String, String>();
	
	/**
	 * 建立ws连接;
	 *
	 * @param session:当前用户打开的会话实例;
	 * @param userId:当前用户的id;
	 */
	@OnOpen
	public void open(Session session, @PathParam(value = "userId") String userId)
	{
		System.out.println("开启websocket连接的用户Id: " + userId);
		
		//存储Websocket的Session到内存中;
		//this.session = session;
		//sessionPool.put(userId, session);
		//sessionIds.put(session.getId(), userId);
		//存储客户端的sessin信息到Redis中;
		//webSocketUtils.saveSessionInfoToRedis(session, userId);
		
		this.session = session;
		String sessionKey = RedisUtils.generateKeyWithTimestamp(userId);
		sessionPool.put(sessionKey, session);
		sessionIds.put(session.getId(), sessionKey);
		
		//存储客户端的sessin信息到Redis中;
		webSocketUtils.saveSessionInfoToRedis(session, sessionKey);
		
		System.out.println("---WS-------IS------OPening---");
	}
	
	/**
	 * 接受前端ws.send("msg")方法推送到后台的消息;
	 *
	 * @param message:发送的消息内容;
	 */
	@OnMessage
	public void onMessage(String message)
	{
		System.out.println("当前发送人sessionid为" + session.getId() + "发送内容为" + message);
	}
	
	/**
	 * 当连接关闭时调用的逻辑;
	 */
	@OnClose
	public void onClose()
	{
		System.out.println("关闭websocket连接！");
		String sessionKey = sessionIds.get(session.getId());
		sessionPool.remove(sessionKey);
		sessionIds.remove(session.getId());
		webSocketUtils.removeSessionInfoToRedis(sessionKey);
	}
	
	/**
	 * 连接建立异常时调用的方法;
	 *
	 * @param session:当前会话;
	 * @param error:异常栈信息;
	 */
	@OnError
	public void onError(Session session, Throwable error)
	{
		System.out.println("websocket连接异常！");
		error.printStackTrace();
	}
	
	/**
	 * 推送消息到指定的用户;
	 *
	 * @param message:发送的消息;
	 * @param sessionKey:用户的session信息的Id;
	 */
	public static void sendMessage(String message, String sessionKey)
	{
		System.out.println("向Id为：" + sessionKey + "的用户发送消息体" + message);
		Session s = sessionPool.get(sessionKey);
		if (s != null)
		{
			try
			{
				s.getBasicRemote().sendText(message);
			}
			catch (IOException e)
			{
				e.printStackTrace();
			}
		}
	}
	
	/**
	 * 广播消息到所有用户;
	 *
	 * @param msg:发送/推送的消息内容;
	 */
	public static void sendAll(String msg)
	{
		for (String key : sessionIds.keySet())
		{
			sendMessage(msg, sessionIds.get(key));
		}
	}
	
	/**
	 * 广播/推送消息到指定用户;
	 *
	 * @param wsMsg:发送/推送的消息内容;
	 */
	public static void sendMsgToSpecifiedRange(WsMsg wsMsg)
	{
		List<String> receiverList = wsMsg.getReceiverList();
		String msg = wsMsg.getMsgBody();
		for (String receiverId : receiverList)
		{
			sendMessage(msg, receiverId);
		}
	}
	
	/**
	 * 获取在线用户总数;
	 *
	 * @return:在线用户总数;
	 */
	public static int getOnlineNum()
	{
		return sessionPool.size();
	}
	
	/**
	 * 获取在线用户总列表字符串形式;
	 *
	 * @return :在线用户userId列表;
	 */
	public static String getOnlineUsers()
	{
		StringBuffer users = new StringBuffer();
		for (String key : sessionIds.keySet())
		{
			users.append(sessionIds.get(key) + ",");
		}
		return users.toString();
	}
	
	/**
	 * 推送消息到客户端;
	 *
	 * @param msgBody:消息主体;
	 * @param userId:当前访问者的Id,查找会话的依据;
	 */
	public static void pushMsgToClient(Object msgBody, String userId)
	{
		String wsMsgTxt = JSONUtil.toJsonStr(msgBody);
		sendMessage(wsMsgTxt, userId);
	}
	
	/**
	  *   发送消息到在线用户;
	 * @param msg:消息主体;
	 */
	public static void sendMsgToOnline(WsMsg msg)
	{
		//从Redis中获取用户建立链接的队列;
		List<QueueMsg> queueMsgList = getRouteKeyFromRedis(msg);
		
		//推送消息到用户建立链接的队列;
		if (queueMsgList.size() == 0) { return; }
		for (QueueMsg queueMsgIns : queueMsgList)
		{
			String msgBody = JSONUtil.toJsonStr(queueMsgIns);
			String routeKey = queueMsgIns.getRouteKey();
			SocketServer.rabbitMqService.sendMsgToSpecificQueue(msgBody, routeKey);
		}
	}
	
	/**
	  * 从redis中获取所有的在线用户的Id;
	 * @return
	 */
	public static List<String> getAllOnlineUserInfos()
	{
		String keyPattern = RedisConstants.SESSION_CACHE_NAME_SPACE + "*";
		List<String> userInfoSet = new ArrayList<String>();
		List<String> reallyUserInfoSet = new ArrayList<String>();
		
		Set<String> sessionKeys = getRedisInfoByKey(keyPattern);
		for (String keyEle : sessionKeys)
		{
			userInfoSet.add(keyEle);
		}
		
		//只保留userId部分;
		for (String userInfoEle : userInfoSet)
		{
			userInfoEle = userInfoEle.replace(RedisConstants.SESSION_CACHE_NAME_SPACE, "");
			reallyUserInfoSet.add(userInfoEle);
		}
		
		return reallyUserInfoSet;
	}
	
	/**
	 * 根据key的匹配正则从Redis中获取所有的key;
	 * @param keyPattern
	 */
	private static Set<String> getRedisInfoByKey(String keyPattern)
	{
		//切换Redis数据库;
		SocketServer.redisUtils.switchRedisDB(SocketServer.redisTemplate, RedisConstants.USERINFO_DB_INDEX);
		Set<String> keys = SocketServer.redisTemplate.keys(keyPattern);
		return keys;
	}
	
	/**
	  * 从Redis中获取用户的队列路由Key;
	 * @param receiverList
	 * @return
	 */
	private static List<QueueMsg> getRouteKeyFromRedis(WsMsg msg)
	{
		List<String> receiverList = msg.getReceiverList();
		String msgBody = msg.getMsgBody();
		List<QueueMsg> queueMsgList = new ArrayList<QueueMsg>();
		
		if (receiverList == null) { return queueMsgList; }
		
		for (String userId : receiverList)
		{
			try
			{
				//从Redis中获取当前用户已经建立的所有链接的信息;
				String keyPatteran = RedisConstants.SESSION_CACHE_NAME_SPACE + userId + ":" + "*";
				List<Object> sessionInfoSet = getMultiRedisInfoByPattern(keyPatteran);
				
				//装配用户的队列消息;
				assembleQueueMsg(msgBody, queueMsgList, sessionInfoSet);
			}
			catch (Exception e)
			{
				e.printStackTrace();
			}
		}
		
		return queueMsgList;
	}
	
	/**
	 * 装配用户的队列消息;
	 * @param msgBody
	 * @param queueMsgList
	 * @param sessionInfoSet
	 */
	public static void assembleQueueMsg(String msgBody, List<QueueMsg> queueMsgList, List<Object> sessionInfoSet)
	{
		for (Object sessionInfoEle : sessionInfoSet)
		{
			SessionInfo sessionInfo = JSONUtil.toBean(sessionInfoEle.toString(), SessionInfo.class);
			String routeKey = sessionInfo.getRouteKey();
			String sessionId = sessionInfo.getSessionId();
			QueueMsg queueMsg = new QueueMsg();
			queueMsg.setSessionId(sessionId);
			queueMsg.setRouteKey(routeKey);
			queueMsg.setMsgBody(msgBody);
			queueMsgList.add(queueMsg);
		}
	}
	
	/**
	 * 根据用户Id从Redis中获取用户信息;
	* @param userId
	* @return
	*/
	public static String getRedisInfoByUserId(String userId)
	{
		//切换Redis数据库;
		SocketServer.redisUtils.switchRedisDB(SocketServer.redisTemplate, RedisConstants.USERINFO_DB_INDEX);
		Object redisMsgIns = SocketServer.redisTemplate.opsForValue().get(RedisConstants.SESSION_CACHE_NAME_SPACE + userId);
		String routeKey = redisMsgIns.toString();
		return routeKey;
	}
	
	/**
	* 根据用户Id从Redis中获取用户信息[Multi];
	* @param userId
	* @return
	*/
	public static List<Object> getMultiRedisInfoByPattern(String pattern)
	{
		//RedisConstants.USERINFO_DB_INDEX
		//切换Redis数据库;
		//SocketServer.redisUtils.switchRedisDB(SocketServer.redisTemplate, dbIndex);
		Set<String> userSessionInfoKeys = SocketServer.redisTemplate.keys(pattern);
		List<Object> userSessionInfoSet = SocketServer.redisTemplate.opsForValue().multiGet(userSessionInfoKeys);
		return userSessionInfoSet;
	}
	
	public static void main(String[] args)
	{
		String sessionKey = RedisUtils.generateKeyWithTimestamp("20142213");
		System.out.println(sessionKey);
	}
}
